package cn.wangjie.spark.continuous

import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
 * 实时从Kafka Topic消费基站日志数据，过滤获取通话转态为success数据，再存储至Kafka Topic中
 * 1、从KafkaTopic中获取基站日志数据（模拟数据，JSON格式数据）
 * 2、ETL：只获取通话状态为success日志数据
 * 3、最终将ETL的数据存储到Kafka Topic中
 */
object StructuredContinuous {
	
	def main(args: Array[String]): Unit = {
		
		// 创建SparkSession实例对象
		val spark: SparkSession = SparkSession
			.builder()
			.appName(this.getClass.getSimpleName.stripSuffix("$"))
			.master("local[2]")
			// 设置SparkSQL Shuffle时分区数目
			.config("spark.sql.shuffle.partitions", "2")
			.getOrCreate()
		// 导入隐式转换和函数库
		import spark.implicits._
		
		// TODO: 1. 从Kafka消费数据
		val kafkaStreamDF: DataFrame = spark
			.readStream
			.format("kafka")
			.option("kafka.bootstrap.servers", "node1.itcast.cn:9092")
			.option("subscribe", "stationTopic")
			.load()
		
		// TODO: 2. 获取每条消息Message
		val etlStreamDF: Dataset[String] = kafkaStreamDF
			// 将value转换为String字符串类型
			.selectExpr("CAST(value AS STRING)")
			// 将DataFrame转换为Dataset
			.as[String]
			// 过滤获取通话转态为success数据
			// 数据样本：station_5,18600000519,18900006900,success,1597568044162,9000
    		.filter{line =>
			    null != line &&
				    line.trim.split(",").length == 6 &&
				    "success".equals(line.trim.split(",")(3))
		    }
		//etlStreamDF.printSchema()
		
		//  输出数据和启动启用
		val query: StreamingQuery = etlStreamDF
			.writeStream
			.outputMode(OutputMode.Append())
			.format("kafka")
			.option("kafka.bootstrap.servers", "node1.itcast.cn:9092")
			.option("topic", "etlTopic")
    		.option("checkpointLocation", "datas/structured/etl-1001")
			// TODO: 设置流处理模式为Continue Processing持续处理模式
    		.trigger(Trigger.Continuous("1 second"))
			.start()
		query.awaitTermination() // 当流式应用运行以后，正常情况一直运行，除非人为或程序异常终止，等待结束
		query.stop()
	}
	
}
